#include "consumer/client/MessageSubscriber.h"
#include "consumer/client/ConsumerMediator.h"
#include "codec/RequestBuilder.h"

namespace zeroplus::mq::consumer
{
    MessageSubscriber::MessageSubscriber(std::shared_ptr<ConsumerMediator> mediator)
        : mediator_(mediator)
    {
    }

    MessageSubscriber::~MessageSubscriber() {}

    transport::SubscribeMessageResponse MessageSubscriber::SendSubscribeMessageRequest(const std::string &topic)
    {
        topic_ = topic;
        transport::SubscribeMessageRequest request;
        request.set_topic(topic);
        request.set_consumer_group("zeroplus_group");
        request.set_consumer_id(Utils::RandomString(5));

        RequestBuilder<transport::RequestCode::SUB_MESSAGE,
                       transport::SubscribeMessageRequest,
                       transport::SubscribeMessageResponse>
            builder(client_);
        auto [responseCode, response] = builder.InvokeSync(request);
        return response;
    }

    void MessageSubscriber::SendAckMessage(const std::string &messageId, const transport::ConsumeStatus &code)
    {
        transport::SendAckMessage request;
        request.set_status(code);
        request.set_topic(topic_);
        request.set_message_id(messageId);
        request.set_consumer_group("zeroplus_group");

        RequestBuilder<transport::RequestCode::SEND_ACK,
                       transport::SendAckMessage,
                       void>
            builder(client_);
        return builder.InvokeOneWay(request);
    }

    void MessageSubscriber::ConnectToBroker(const std::string &uri)
    {
        Uri uriParser = Uri(uri);
        client_ = std::make_unique<ClientNode>(uriParser.GetUri());
        subscriber_ = std::make_unique<SubscriberNode>(uriParser.GetNextUri(PUSHMODE));
    }

    void MessageSubscriber::SubscribeMessage(const std::string &topic)
    {
        assert(client_ != nullptr && subscriber_ != nullptr);
        subscriber_->Subscribe(topic);
        LOG_DEBUG("发送消息订阅请求");
        transport::SubscribeMessageResponse subResponse = SendSubscribeMessageRequest(topic);
    }

    bool MessageSubscriber::Listen(const std::function<void(const Bytes &)> OnListeningCb)
    {
        assert(subscriber_ != nullptr);
        if (subscriber_->isListened_)
        {
            LOG_FATAL("Has listend.");
            return false;
        }
        bool res = subscriber_->Run(OnListeningCb);
        subscriber_->isListened_ = res;
        return res;
    }
}
